package cn.com.sparkle.paxos.client;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.client.PaxosOperater.WatchCallBack;
import cn.com.sparkle.paxos.deamon.ReConnectThread;
import cn.com.sparkle.paxos.deamon.ReConnectThread.ReConnectMethod;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.net.NetNode;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.CustomOnlyConnectMaster;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.raptor.core.buff.SyncBuffPool;
import cn.com.sparkle.raptor.core.handler.IoHandler;
import cn.com.sparkle.raptor.core.protocol.MultiThreadProtecolHandler.ProtocolHandlerIoSession;
import cn.com.sparkle.raptor.core.protocol.Protocol;
import cn.com.sparkle.raptor.core.protocol.ProtocolHandler;
import cn.com.sparkle.raptor.core.transport.socket.nio.IoSession;
import cn.com.sparkle.raptor.core.transport.socket.nio.NioSocketClient;
import cn.com.sparkle.raptor.core.transport.socket.nio.exception.SessionHavaClosedException;

public class ClientHandler implements ProtocolHandler {

	private final static Logger logger = Logger.getLogger(ClientHandler.class);

	private IoHandler handler;

	private NioSocketClient client;
	private ReConnectThread reConnectThread;
	private EventsManager eventsManager;
	
	private AtomicLong watchId = new AtomicLong(0);
	private ConcurrentHashMap<Long, WatchCallBack> watchCommand = new ConcurrentHashMap<Long, WatchCallBack>();
	
	public long register(WatchCallBack callback){
		long id;
		for(;;){
			long oldId  = watchId.get();
			id = 0;
			if(oldId != Long.MAX_VALUE){
				id = oldId + 1;
			}
			if(watchId.compareAndSet(oldId, id)){
				if(!watchCommand.containsKey(id)){
					break;
				}
			}
		}
		watchCommand.put(id, callback);
		return id;
	}
	
	private final static class ClientAttachment {
		private ConnectConfig connectConfig;
		private NetNode nNode;

		public ClientAttachment(ConnectConfig connectConfig, NetNode nNode) {
			super();
			this.connectConfig = connectConfig;
			this.nNode = nNode;
		}

		public ConnectConfig getConnectConfig() {
			return connectConfig;
		}

		public NetNode getnNode() {
			return nNode;
		}
	}

	public ClientHandler(NioSocketClient client, ReConnectThread reConnectThread,EventsManager eventsManager) {
		this.client = client;
		this.reConnectThread = reConnectThread;
		this.eventsManager = eventsManager;
	}

	public void setHandler(IoHandler handler) {
		this.handler = handler;
	}

	@Override
	public void onOneThreadSessionOpen(ProtocolHandlerIoSession session) {

		// String address = (String) session.customAttachment;
		ConnectConfig connectConfig = (ConnectConfig) session.customAttachment;
		logger.debug("connected " + connectConfig.getAddress());
		NetNode nNode = new NetNode(session, new NodeState(
				connectConfig.getAddress()));
		ClientAttachment clientAttachment = new ClientAttachment(connectConfig,
				nNode);
		session.customAttachment = clientAttachment;

		if (clientAttachment.getConnectConfig().isKeepMaster()) {
			// write keep master
			try {
				session.writeObject(CustomOnlyConnectMaster
						.getDefaultInstance());
			} catch (SessionHavaClosedException e) {
				session.closeSession();
				return;
			}
		}
		connectConfig.connected(nNode);
	}

	@Override
	public void onOneThreadSessionClose(ProtocolHandlerIoSession session) {
		ClientAttachment clientAttachment = (ClientAttachment) session.customAttachment;
		// NetNode node = (NetNode) session.customAttachment;
		NetNode node = clientAttachment.getnNode();
		clientAttachment.getConnectConfig().disconnected(node);
		// processor.disconnected(node);
		reConnect(clientAttachment.getConnectConfig());
	}

	@Override
	public void onOneThreadCatchException(IoSession session,
			ProtocolHandlerIoSession attachment, Throwable e) {
		logger.error(session.getRemoteAddress(), e);
		if (attachment == null) {
			reConnect((ConnectConfig) session.attachment());
		}

	}

	@Override
	public void onOneThreadMessageRecieved(Object receiveObject,
			ProtocolHandlerIoSession session) {
		if (receiveObject instanceof MessagePackage) {
			ClientAttachment clientAttachment = (ClientAttachment) session.customAttachment;
			// NetNode n = (NetNode) session.customAttachment;
			clientAttachment.getnNode().recieve((MessagePackage) receiveObject);
		}
	}

	@Override
	public void onOneThreadMessageSent(ProtocolHandlerIoSession attachment,
			int sendSize) {
		// nothing to do
	}

	private final class ReConnect implements ReConnectMethod {
		private final Logger log = Logger.getLogger(ReConnect.class);

		@Override
		public void reConnect(Object value) {
			ConnectConfig connectConfig = (ConnectConfig) value;
			// String address = (String) value;
			String[] a = connectConfig.getAddress().split(":");
			try {
				log.debug("reConnect " + connectConfig.getAddress());
				Future<Boolean> future = client.connect(new InetSocketAddress(
						a[0], Integer.parseInt(a[1])), handler, connectConfig);
				/*
				 * boolean isSuccess = future.get(); if(isSuccess){ //advice to
				 * change this node }
				 */
			} catch (Exception e) {
				reConnectThread.add(connectConfig, this, 2000);
			}
		}

	}

	private final ReConnect method = new ReConnect();

	private void reConnect(ConnectConfig connectConfig) {
		// After sleep 5 seconds,retry to connect.
		if (connectConfig.isAutoReConnect()) {
			logger.debug("reconnect " + connectConfig.getAddress());
			reConnectThread.add(connectConfig, method, 5000);
		}
	}
}
